package com.cw.realtime.common
package util

import constant.Constant._

import org.apache.doris.flink.cfg.{DorisExecutionOptions, DorisOptions, DorisReadOptions}
import org.apache.doris.flink.sink.DorisSink
import org.apache.doris.flink.sink.writer.SimpleStringSerializer
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.base.DeliveryGuarantee
import org.apache.flink.connector.kafka.sink.{KafkaRecordSerializationSchemaBuilder, KafkaSink}

import java.util.Properties

object FlinkSinkUtil {

  def getKafkaSink(topicName: String): KafkaSink[String] = {
    val serializer = new
        KafkaRecordSerializationSchemaBuilder[String]
      .setTopic(topicName)
      .setValueSerializationSchema(new SimpleStringSchema())
      .build()

    KafkaSink.builder()
      .setBootstrapServers(KAFKA_BROKERS)
      .setRecordSerializer(serializer)
      .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
      .setTransactionalIdPrefix(s"cw-$topicName" + System.currentTimeMillis())
      .setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "")
      .build()

  }

  def getDorisSink(tableName: String) = {

    val properties = new Properties()
    properties.setProperty("format", "json");
    properties.setProperty("read_json_by_line", "true");

    DorisSink
      .builder[String]
      // 配置连接
      .setDorisOptions(
        DorisOptions
          .builder
          .setFenodes(FENODES)
          .setTableIdentifier(s"$DORIS_DB.$tableName")
          .setUsername(DORIS_USERNAME)
          .setPassword(DORIS_PASSWD)
          .build()
      )
      .setDorisReadOptions(DorisReadOptions.builder.build())
      // 配置执行
      .setDorisExecutionOptions(
        DorisExecutionOptions
          .builder
          .setLabelPrefix("label-doris" + System.currentTimeMillis)
          .setDeletable(false)
          .setStreamLoadProp(properties)
          .build()
      )
      .setSerializer(new SimpleStringSerializer())
      .build()

  }



}
